package cn.flying.cloud.task.core.batch.listener;

import javax.annotation.Resource;

import java.time.Duration;
import java.time.LocalDateTime;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.scope.context.StepContext;
import org.springframework.stereotype.Component;

import cn.flying.cloud.base.common.utils.Ft;
import cn.flying.cloud.task.core.callback.disruptor.AsyncJobLogDisruptor;
import cn.flying.cloud.task.core.model.StepResult;


/**
 * 批量执行监控
 * 执行顺序：beforeJob -> beforeStep -> beforeChunk -> afterChunk/afterChunkError（提交事务出错走error） -> afterStep（无论失败与否都会执行） -> afterJob（无论失败与否都会执行）
 *
 * @author: admin
 * @date: 2024年01月12日 14:50
 * @version: 1.0
 */
@Component
public class BatchMonitorExecutionListener implements ChunkListener, StepExecutionListener, JobExecutionListener {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private static ThreadLocal<StepResult> RESULT_THREAD_LOCAL = ThreadLocal.withInitial(StepResult::new);

    @Resource
    private AsyncJobLogDisruptor asyncJobLogDisruptor;

    /**
     * Callback before the chunk is executed, but inside the transaction.
     *
     * @param chunkContext The current {@link ChunkContext}
     */
    @Override
    public void beforeChunk(ChunkContext chunkContext) {
        StepContext stepContext = chunkContext.getStepContext();
        logger.info("批量任务：【{}】实例ID：【{}】步骤：【{}】提交事务开始！时间：【{}】", stepContext.getJobName(), stepContext.getJobInstanceId(), stepContext.getStepName(), LocalDateTime.now());
    }

    /**
     * Callback after the chunk is executed, outside the transaction.
     *
     * @param chunkContext The current {@link ChunkContext}
     */
    @Override
    public void afterChunk(ChunkContext chunkContext) {
        StepContext stepContext = chunkContext.getStepContext();
        logger.info("批量任务：【{}】实例ID：【{}】步骤：【{}】提交事务完成！时间：【{}】", stepContext.getJobName(), stepContext.getJobInstanceId(), stepContext.getStepName(), LocalDateTime.now());
    }

    /**
     * Callback after a chunk has been marked for rollback.  It is invoked
     * after transaction rollback.  While the rollback will have occurred,
     * transactional resources might still be active and accessible.  Due to
     * this, data access code within this callback will still "participate" in
     * the original transaction unless it declares that it runs in its own
     * transaction.  Hence: <em> Use PROPAGATION_REQUIRES_NEW for any
     * transactional operation that is called from here.</em>
     *
     * @param chunkContext the chunk chunkContext containing the exception that caused
     *                     the underlying rollback.
     */
    @Override
    public void afterChunkError(ChunkContext chunkContext) {
        StepContext stepContext = chunkContext.getStepContext();
        logger.warn("批量任务：【{}】实例ID：【{}】步骤：【{}】提交事务出错了！时间：【{}】", stepContext.getJobName(), stepContext.getJobInstanceId(), stepContext.getStepName(), LocalDateTime.now());
    }

    /**
     * Initialize the state of the listener with the {@link StepExecution} from
     * the current scope.
     *
     * @param stepExecution instance of {@link StepExecution}.
     */
    @Override
    public void beforeStep(StepExecution stepExecution) {
        JobExecution jobExecution = stepExecution.getJobExecution();
        JobInstance jobInstance = jobExecution.getJobInstance();
        logger.info("批量任务：【{}】实例ID：【{}】步骤：【{}】执行开始！时间：【{}】", jobInstance.getJobName(), jobInstance.getId(), stepExecution.getStepName(), LocalDateTime.now());

        StepResult result = RESULT_THREAD_LOCAL.get();
        result.setSerialNo(jobExecution.getJobParameters().getString("serialNo"));
        result.setJobName(jobInstance.getJobName());
        result.setJobId(jobInstance.getId());
        result.setExecutionId(stepExecution.getJobExecutionId());
        result.setStepId(stepExecution.getId());
        result.setStepName(stepExecution.getStepName());
        result.setStepDesc("");
        result.setStartTime(LocalDateTime.now());
        result.setTenantId(jobExecution.getJobParameters().getString("tenantId"));
    }

    /**
     * Give a listener a chance to modify the exit status from a step. The value
     * returned will be combined with the normal exit status using
     * {@link ExitStatus#and(ExitStatus)}.
     * <p>
     * Called after stepExecution of step's processing logic (both successful or
     * failed). Throwing exception in this method has no effect, it will only be
     * logged.
     *
     * @param stepExecution {@link StepExecution} instance.
     * @return an {@link ExitStatus} to combine with the normal value. Return
     * {@code null} to leave the old value unchanged.
     */
    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        JobExecution jobExecution = stepExecution.getJobExecution();
        JobInstance jobInstance = jobExecution.getJobInstance();
        logger.info("批量任务：【{}】实例ID：【{}】步骤：【{}】执行结束！时间：【{}】，状态信息：【{}】", jobInstance.getJobName(), jobInstance.getId(), stepExecution.getStepName(), LocalDateTime.now(), stepExecution.getExitStatus());

        StepResult result = RESULT_THREAD_LOCAL.get();

        result.setEndTime(LocalDateTime.now());
        result.setStatus(stepExecution.getExitStatus().getExitCode());
        result.setStepMsg("执行成功！");
        if (stepExecution.getExitStatus().getExitCode().equals(ExitStatus.FAILED.getExitCode())) {
            result.setStepMsg(stepExecution.getExitStatus().getExitCode());
            result.setStepTrace(stepExecution.getExitStatus().getExitDescription());
        }
        Duration duration = Duration.between(result.getStartTime(), result.getEndTime());
        result.setDuration(duration.getSeconds());

        asyncJobLogDisruptor.publish("step", Ft.JSON.serializer(result));
        return stepExecution.getExitStatus();
    }

    /**
     * Callback before a job executes.
     *
     * @param jobExecution the current {@link JobExecution}
     */
    @Override
    public void beforeJob(JobExecution jobExecution) {
        logger.info("批量任务：【{}】执行开始！实例ID：【{}】时间：【{}】", jobExecution.getJobInstance().getJobName(), jobExecution.getJobInstance().getId(), LocalDateTime.now());
    }

    /**
     * Callback after completion of a job. Called after both both successful and
     * failed executions. To perform logic on a particular status, use
     * "if (jobExecution.getStatus() == BatchStatus.X)".
     *
     * @param jobExecution the current {@link JobExecution}
     */
    @Override
    public void afterJob(JobExecution jobExecution) {
        logger.info("批量任务：【{}】执行结束！实例ID：【{}】时间：【{}】", jobExecution.getJobInstance().getJobName(), jobExecution.getJobInstance().getId(), LocalDateTime.now());
    }


}